--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 3a9f6eb31667cd0b8e4d1d672bf1c84a85bc31fb
Parents : 2be0198
Author : Sudo-Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-03-05T15:59:47-06:00
Add tests for database backup logic and interface discovery configuration
Changes
7 files changed, 1167 insertions(+), 2 deletions(-)
Diff
diff --git a/tests/backend/test_database_snapshots.py b/tests/backend/test_database_snapshots.py
index bdcd0fd5..13eca3ca 100644
--- a/tests/backend/test_database_snapshots.py
+++ b/tests/backend/test_database_snapshots.py
@@ -1,3 +1,4 @@
+import json
import os
import shutil
import tempfile
@@ -75,4 +76,192 @@ def test_database_auto_backup_logic(temp_dir):
assert os.path.exists(result["path"])
backup_dir = os.path.join(temp_dir, "database-backups")
- assert len(os.listdir(backup_dir)) == 1
+ zips = [f for f in os.listdir(backup_dir) if f.endswith(".zip")]
+ assert len(zips) == 1
+
+
+def test_backup_baseline_created_on_first_backup(temp_dir):
+ db_path = os.path.join(temp_dir, "test.db")
+ db = Database(db_path)
+ db.initialize()
+ db.backup_database(temp_dir)
+ baseline_path = os.path.join(temp_dir, "database-backups", "backup-baseline.json")
+ assert os.path.exists(baseline_path)
+ with open(baseline_path) as f:
+ data = json.load(f)
+ assert "message_count" in data
+ assert "total_bytes" in data
+ assert "timestamp" in data
+
+
+def test_backup_suspicious_when_messages_gone_skips_cleanup_and_baseline(temp_dir):
+ db_path = os.path.join(temp_dir, "test.db")
+ db = Database(db_path)
+ db.initialize()
+ db.messages.upsert_lxmf_message({
+ "hash": "h1", "source_hash": "s", "destination_hash": "d", "peer_hash": "p",
+ "state": "delivered", "progress": 1.0, "is_incoming": 1, "method": "direct",
+ "delivery_attempts": 0, "next_delivery_attempt_at": None, "title": "t", "content": "c",
+ "fields": "{}", "timestamp": 0, "rssi": None, "snr": None, "quality": None,
+ "is_spam": 0, "reply_to_hash": None,
+ })
+ result1 = db.backup_database(temp_dir, max_count=3)
+ assert result1.get("suspicious") is not True
+ backup_dir = os.path.join(temp_dir, "database-backups")
+ zip_count_after_first = sum(1 for f in os.listdir(backup_dir) if f.endswith(".zip"))
+ assert zip_count_after_first == 1
+ with open(os.path.join(backup_dir, "backup-baseline.json")) as f:
+ baseline_after_first = json.load(f)
+ assert baseline_after_first["message_count"] == 1
+
+ db.messages.delete_all_lxmf_messages()
+ assert db.messages.count_lxmf_messages() == 0
+ result2 = db.backup_database(temp_dir, max_count=3)
+ assert result2.get("suspicious") is True
+ assert "baseline" in result2
+ assert result2["baseline"]["message_count"] == 1
+ assert result2["current_stats"]["message_count"] == 0
+ zip_count_after_suspicious = sum(1 for f in os.listdir(backup_dir) if f.endswith(".zip"))
+ assert zip_count_after_suspicious == 2
+ assert any("SUSPICIOUS" in f for f in os.listdir(backup_dir) if f.endswith(".zip"))
+ with open(os.path.join(backup_dir, "backup-baseline.json")) as f:
+ baseline_after_suspicious = json.load(f)
+ assert baseline_after_suspicious["message_count"] == 1
+
+
+def test_backup_suspicious_when_size_collapsed_skips_cleanup(temp_dir):
+ db_path = os.path.join(temp_dir, "test.db")
+ db = Database(db_path)
+ db.initialize()
+ for i in range(200):
+ db.messages.upsert_lxmf_message({
+ "hash": f"h{i}", "source_hash": "s", "destination_hash": "d", "peer_hash": "p",
+ "state": "delivered", "progress": 1.0, "is_incoming": 1, "method": "direct",
+ "delivery_attempts": 0, "next_delivery_attempt_at": None, "title": "t", "content": "x" * 500,
+ "fields": "{}", "timestamp": float(i), "rssi": None, "snr": None, "quality": None,
+ "is_spam": 0, "reply_to_hash": None,
+ })
+ db.backup_database(temp_dir, max_count=3)
+ baseline_path = os.path.join(temp_dir, "database-backups", "backup-baseline.json")
+ with open(baseline_path) as f:
+ baseline = json.load(f)
+ assert baseline["total_bytes"] > 100_000
+ db.messages.delete_all_lxmf_messages()
+ db.execute_sql("VACUUM")
+ result = db.backup_database(temp_dir, max_count=3)
+ assert result.get("suspicious") is True
+ backup_dir = os.path.join(temp_dir, "database-backups")
+ zips = [f for f in os.listdir(backup_dir) if f.endswith(".zip")]
+ assert len(zips) >= 2
+
+
+def test_backup_normal_rotation_and_baseline_update(temp_dir):
+ import time
+ db_path = os.path.join(temp_dir, "test.db")
+ db = Database(db_path)
+ db.initialize()
+ db.backup_database(temp_dir, max_count=2)
+ time.sleep(1.1)
+ db.backup_database(temp_dir, max_count=2)
+ time.sleep(1.1)
+ db.backup_database(temp_dir, max_count=2)
+ backup_dir = os.path.join(temp_dir, "database-backups")
+ zips = sorted([f for f in os.listdir(backup_dir) if f.endswith(".zip")])
+ assert len(zips) == 2
+ assert os.path.exists(os.path.join(backup_dir, "backup-baseline.json"))
+
+
+def test_backup_failure_does_not_remove_existing_backups(temp_dir):
+ from unittest.mock import patch
+ db_path = os.path.join(temp_dir, "test.db")
+ db = Database(db_path)
+ db.initialize()
+ db.backup_database(temp_dir)
+ backup_dir = os.path.join(temp_dir, "database-backups")
+ existing = [f for f in os.listdir(backup_dir) if f.endswith(".zip")]
+ assert len(existing) == 1
+ with patch.object(db, "_backup_to_zip", side_effect=OSError("disk full")):
+ with pytest.raises(OSError):
+ db.backup_database(temp_dir, max_count=1)
+ still_there = [f for f in os.listdir(backup_dir) if f.endswith(".zip")]
+ assert len(still_there) == 1
+
+
+def test_check_db_health_at_open_no_baseline_ok(temp_dir):
+ db_path = os.path.join(temp_dir, "test.db")
+ db = Database(db_path)
+ db.initialize()
+ issues = db.check_db_health_at_open(temp_dir)
+ assert issues == []
+
+
+def test_check_db_health_at_open_baseline_suspicious_content(temp_dir):
+ db_path = os.path.join(temp_dir, "test.db")
+ db = Database(db_path)
+ db.initialize()
+ db.messages.upsert_lxmf_message({
+ "hash": "h1", "source_hash": "s", "destination_hash": "d", "peer_hash": "p",
+ "state": "delivered", "progress": 1.0, "is_incoming": 1, "method": "direct",
+ "delivery_attempts": 0, "next_delivery_attempt_at": None, "title": "t", "content": "c",
+ "fields": "{}", "timestamp": 0, "rssi": None, "snr": None, "quality": None,
+ "is_spam": 0, "reply_to_hash": None,
+ })
+ db.backup_database(temp_dir)
+ db.messages.delete_all_lxmf_messages()
+ issues = db.check_db_health_at_open(temp_dir)
+ assert len(issues) >= 1
+ assert any("anomaly" in i.lower() or "messages" in i for i in issues)
+
+
+def test_check_db_health_at_open_integrity_fail(temp_dir):
+ from unittest.mock import patch
+ db_path = os.path.join(temp_dir, "test.db")
+ db = Database(db_path)
+ db.initialize()
+ with patch.object(db.provider, "integrity_check", return_value=[("corrupt",)]):
+ issues = db.check_db_health_at_open(temp_dir)
+ assert len(issues) >= 1
+ assert any("integrity" in i.lower() for i in issues)
+
+
+def test_check_db_health_at_close_no_issues(temp_dir):
+ db_path = os.path.join(temp_dir, "test.db")
+ db = Database(db_path)
+ db.initialize()
+ issues = db.check_db_health_at_close(temp_dir)
+ assert issues == []
+
+
+def test_check_db_health_at_close_integrity_fail(temp_dir):
+ from unittest.mock import patch
+ db_path = os.path.join(temp_dir, "test.db")
+ db = Database(db_path)
+ db.initialize()
+ with patch.object(db.provider, "integrity_check", return_value=[("corrupt",)]):
+ issues = db.check_db_health_at_close(temp_dir)
+ assert len(issues) >= 1
+ assert any("integrity" in i.lower() for i in issues)
+
+
+def test_is_backup_suspicious_does_not_mistrigger_empty_baseline():
+ from meshchatx.src.backend.database import Database
+ db = Database(":memory:")
+ db.initialize()
+ assert db._is_backup_suspicious({"message_count": 0, "total_bytes": 0}, None) is False
+ assert db._is_backup_suspicious({"message_count": 10, "total_bytes": 1000}, None) is False
+
+
+def test_is_backup_suspicious_does_not_mistrigger_legitimate_empty():
+ from meshchatx.src.backend.database import Database
+ db = Database(":memory:")
+ db.initialize()
+ baseline = {"message_count": 0, "total_bytes": 5000}
+ assert db._is_backup_suspicious({"message_count": 0, "total_bytes": 5000}, baseline) is False
+
+
+def test_is_backup_suspicious_does_not_mistrigger_small_db():
+ from meshchatx.src.backend.database import Database
+ db = Database(":memory:")
+ db.initialize()
+ baseline = {"message_count": 5, "total_bytes": 50_000}
+ assert db._is_backup_suspicious({"message_count": 5, "total_bytes": 55_000}, baseline) is False
diff --git a/tests/backend/test_https_wss_side_sniffing.py b/tests/backend/test_https_wss_side_sniffing.py
new file mode 100644
index 00000000..6c8d2c4d
--- /dev/null
+++ b/tests/backend/test_https_wss_side_sniffing.py
@@ -0,0 +1,168 @@
+"""
+Tests that HTTPS/WSS is used so local traffic cannot be sniffed by other apps.
+Server must speak TLS only on the API/WS port; plain HTTP must not be accepted.
+"""
+
+import asyncio
+import os
+import socket
+import ssl
+import tempfile
+from datetime import UTC, datetime, timedelta
+
+import aiohttp
+import pytest
+from aiohttp import web
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import hashes, serialization
+from cryptography.hazmat.primitives.asymmetric import rsa
+from cryptography.x509.oid import NameOID
+
+
+def _make_self_signed_cert_and_key(cert_path: str, key_path: str) -> None:
+ private_key = rsa.generate_private_key(
+ public_exponent=65537,
+ key_size=2048,
+ backend=default_backend(),
+ )
+ subject = issuer = x509.Name(
+ [
+ x509.NameAttribute(NameOID.COUNTRY_NAME, "US"),
+ x509.NameAttribute(NameOID.COMMON_NAME, "localhost"),
+ ],
+ )
+ cert = (
+ x509.CertificateBuilder()
+ .subject_name(subject)
+ .issuer_name(issuer)
+ .public_key(private_key.public_key())
+ .serial_number(x509.random_serial_number())
+ .not_valid_before(datetime.now(UTC))
+ .not_valid_after(datetime.now(UTC) + timedelta(days=365))
+ .sign(private_key, hashes.SHA256(), default_backend())
+ )
+ os.makedirs(os.path.dirname(cert_path) or ".", exist_ok=True)
+ with open(cert_path, "wb") as f:
+ f.write(cert.public_bytes(serialization.Encoding.PEM))
+ with open(key_path, "wb") as f:
+ f.write(
+ private_key.private_bytes(
+ encoding=serialization.Encoding.PEM,
+ format=serialization.PrivateFormat.PKCS8,
+ encryption_algorithm=serialization.NoEncryption(),
+ ),
+ )
+
+
+@pytest.fixture
+def temp_storage():
+ with tempfile.TemporaryDirectory() as d:
+ yield d
+
+
+@pytest.fixture
+def ssl_context_and_cert(temp_storage):
+ cert_dir = temp_storage
+ cert_path = os.path.join(cert_dir, "cert.pem")
+ key_path = os.path.join(cert_dir, "key.pem")
+ _make_self_signed_cert_and_key(cert_path, key_path)
+ ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+ ctx.load_cert_chain(cert_path, key_path)
+ return ctx, cert_path, key_path
+
+
+@pytest.mark.asyncio
+async def test_https_serves_over_tls_only_plain_http_gets_no_http_response(
+ ssl_context_and_cert,
+):
+ """
+ Server started with ssl_context must not respond to plain HTTP.
+ A local sniffer connecting with raw TCP and sending HTTP would get
+ TLS handshake bytes or connection closure, not plaintext HTTP response.
+ """
+ ssl_context, _, _ = ssl_context_and_cert
+ app = web.Application()
+
+ async def root_handler(_request):
+ return web.Response(text="ok")
+
+ app.router.add_get("/", root_handler)
+ runner = web.AppRunner(app, keepalive_timeout=0)
+ await runner.setup()
+ site = web.TCPSite(runner, "127.0.0.1", 0, ssl_context=ssl_context)
+ await site.start()
+ try:
+ port = site._server.sockets[0].getsockname()[1]
+ client_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+ client_ctx.check_hostname = False
+ client_ctx.verify_mode = ssl.CERT_NONE
+
+ async with aiohttp.ClientSession() as session:
+ resp = await session.get(
+ f"https://127.0.0.1:{port}/",
+ ssl=client_ctx,
+ )
+ assert resp.status == 200
+ assert (await resp.text()) == "ok"
+
+ sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ sock.settimeout(1.0)
+ try:
+ sock.connect(("127.0.0.1", port))
+ sock.sendall(b"GET / HTTP/1.0\r\n\r\n")
+ try:
+ raw = sock.recv(1024)
+ except socket.timeout:
+ raw = b""
+ finally:
+ sock.close()
+ assert not raw.startswith(b"HTTP/"), (
+ "Server must not respond with plain HTTP when TLS is enabled; "
+ "plaintext would allow local side-sniffing"
+ )
+ finally:
+ await runner.cleanup()
+
+
+@pytest.mark.asyncio
+async def test_wss_over_same_tls_port(ssl_context_and_cert):
+ """
+ WebSocket upgrade over the same TLS port uses WSS (encrypted).
+ Verifies that a WS endpoint is reachable only via TLS.
+ """
+ ssl_context, _, _ = ssl_context_and_cert
+ app = web.Application()
+
+ async def ws_handler(request):
+ ws = web.WebSocketResponse()
+ await ws.prepare(request)
+ await ws.close()
+ return ws
+
+ app.router.add_get("/ws", ws_handler)
+ runner = web.AppRunner(app, keepalive_timeout=0)
+ await runner.setup()
+ site = web.TCPSite(runner, "127.0.0.1", 0, ssl_context=ssl_context)
+ await site.start()
+ try:
+ port = site._server.sockets[0].getsockname()[1]
+ client_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+ client_ctx.check_hostname = False
+ client_ctx.verify_mode = ssl.CERT_NONE
+
+ async with aiohttp.ClientSession() as session:
+ async with session.ws_connect(
+ f"wss://127.0.0.1:{port}/ws",
+ ssl=client_ctx,
+ ) as ws:
+ msg = await ws.receive()
+ assert msg.type in (
+ aiohttp.WSMsgType.CLOSE,
+ aiohttp.WSMsgType.CLOSING,
+ aiohttp.WSMsgType.CLOSED,
+ )
+ finally:
+ await runner.cleanup()
+
+
diff --git a/tests/backend/test_interface_discovery.py b/tests/backend/test_interface_discovery.py
index ab919a75..a96ca8b4 100644
--- a/tests/backend/test_interface_discovery.py
+++ b/tests/backend/test_interface_discovery.py
@@ -50,6 +50,8 @@ async def test_reticulum_discovery_get_and_patch(temp_dir):
"reticulum": {
"discover_interfaces": "true",
"interface_discovery_sources": "abc,def",
+ "interface_discovery_whitelist": "tcp-*,10.0.*",
+ "interface_discovery_blacklist": "tcp-bad,*:9999",
"required_discovery_value": "16",
"autoconnect_discovered_interfaces": "2",
"network_identity": "/tmp/net_id",
@@ -93,6 +95,8 @@ async def test_reticulum_discovery_get_and_patch(temp_dir):
get_data = json.loads(get_response.body)
assert get_data["discovery"]["discover_interfaces"] == "true"
assert get_data["discovery"]["interface_discovery_sources"] == "abc,def"
+ assert get_data["discovery"]["interface_discovery_whitelist"] == "tcp-*,10.0.*"
+ assert get_data["discovery"]["interface_discovery_blacklist"] == "tcp-bad,*:9999"
assert get_data["discovery"]["required_discovery_value"] == "16"
assert get_data["discovery"]["autoconnect_discovered_interfaces"] == "2"
assert get_data["discovery"]["network_identity"] == "/tmp/net_id"
@@ -101,6 +105,8 @@ async def test_reticulum_discovery_get_and_patch(temp_dir):
new_config = {
"discover_interfaces": False,
"interface_discovery_sources": "",
+ "interface_discovery_whitelist": "peer-*,172.16.*",
+ "interface_discovery_blacklist": "",
"required_discovery_value": 18,
"autoconnect_discovered_interfaces": 5,
"network_identity": "/tmp/other_id",
@@ -115,17 +121,154 @@ async def test_reticulum_discovery_get_and_patch(temp_dir):
patch_data = json.loads(patch_response.body)
assert patch_data["discovery"]["discover_interfaces"] is False
assert patch_data["discovery"]["interface_discovery_sources"] is None
+ assert patch_data["discovery"]["interface_discovery_whitelist"] == "peer-*,172.16.*"
+ assert patch_data["discovery"]["interface_discovery_blacklist"] is None
assert patch_data["discovery"]["required_discovery_value"] == 18
assert patch_data["discovery"]["autoconnect_discovered_interfaces"] == 5
assert patch_data["discovery"]["network_identity"] == "/tmp/other_id"
assert config["reticulum"]["discover_interfaces"] is False
assert "interface_discovery_sources" not in config["reticulum"]
+ assert config["reticulum"]["interface_discovery_whitelist"] == "peer-*,172.16.*"
+ assert "interface_discovery_blacklist" not in config["reticulum"]
assert config["reticulum"]["required_discovery_value"] == 18
assert config["reticulum"]["autoconnect_discovered_interfaces"] == 5
assert config["reticulum"]["network_identity"] == "/tmp/other_id"
assert config.write_called
+@pytest.mark.asyncio
+async def test_discovered_interfaces_respect_whitelist_and_blacklist(temp_dir):
+ config = ConfigDict(
+ {
+ "reticulum": {
+ "interface_discovery_whitelist": "peer-*,10.0.*",
+ "interface_discovery_blacklist": "*:9999,bad-*",
+ },
+ "interfaces": {},
+ },
+ )
+
+ with (
+ patch("meshchatx.meshchat.generate_ssl_certificate"),
+ patch("RNS.Reticulum") as mock_rns,
+ patch("RNS.Transport"),
+ patch("LXMF.LXMRouter"),
+ patch("meshchatx.meshchat.InterfaceDiscovery") as mock_discovery_cls,
+ ):
+ mock_reticulum = mock_rns.return_value
+ mock_reticulum.config = config
+ mock_reticulum.configpath = "/tmp/mock_config"
+ mock_reticulum.is_connected_to_shared_instance = False
+ mock_reticulum.transport_enabled.return_value = True
+ mock_reticulum.get_interface_stats.return_value = {"interfaces": []}
+
+ mock_discovery = mock_discovery_cls.return_value
+ mock_discovery.list_discovered_interfaces.return_value = [
+ {
+ "name": "peer-good-1",
+ "type": "TCPClientInterface",
+ "reachable_on": "10.0.0.7",
+ "port": 4242,
+ },
+ {
+ "name": "peer-blocked-port",
+ "type": "TCPClientInterface",
+ "reachable_on": "10.0.0.8",
+ "port": 9999,
+ },
+ {
+ "name": "bad-node",
+ "type": "TCPClientInterface",
+ "reachable_on": "10.0.0.9",
+ "port": 4242,
+ },
+ {
+ "name": "other-network",
+ "type": "TCPClientInterface",
+ "reachable_on": "192.168.1.10",
+ "port": 4242,
+ },
+ ]
+
+ app_instance = ReticulumMeshChat(
+ identity=build_identity(),
+ storage_dir=temp_dir,
+ reticulum_config_dir=temp_dir,
+ )
+
+ handler = await find_route_handler(
+ app_instance,
+ "/api/v1/reticulum/discovered-interfaces",
+ "GET",
+ )
+ assert handler
+
+ response = await handler(MagicMock())
+ data = json.loads(response.body)
+ interfaces = data["interfaces"]
+
+ assert len(interfaces) == 1
+ assert interfaces[0]["name"] == "peer-good-1"
+
+
+@pytest.mark.asyncio
+async def test_discovery_patch_sanitizes_whitelist_blacklist_values(temp_dir):
+ config = ConfigDict({"reticulum": {}, "interfaces": {}})
+
+ with (
+ patch("meshchatx.meshchat.generate_ssl_certificate"),
+ patch("RNS.Reticulum") as mock_rns,
+ patch("RNS.Transport"),
+ patch("LXMF.LXMRouter"),
+ ):
+ mock_reticulum = mock_rns.return_value
+ mock_reticulum.config = config
+ mock_reticulum.configpath = "/tmp/mock_config"
+ mock_reticulum.is_connected_to_shared_instance = False
+ mock_reticulum.transport_enabled.return_value = True
+
+ app_instance = ReticulumMeshChat(
+ identity=build_identity(),
+ storage_dir=temp_dir,
+ reticulum_config_dir=temp_dir,
+ )
+
+ patch_handler = await find_route_handler(
+ app_instance,
+ "/api/v1/reticulum/discovery",
+ "PATCH",
+ )
+ assert patch_handler
+
+ payload = {
+ "interface_discovery_whitelist": "peer-1,\npeer-1,host:4242,\r\n,\n",
+ "interface_discovery_blacklist": ["bad-node", "bad-node", "evil,\nentry"],
+ }
+
+ class PatchRequest:
+ @staticmethod
+ async def json():
+ return payload
+
+ response = await patch_handler(PatchRequest())
+ data = json.loads(response.body)
+
+ assert data["discovery"]["interface_discovery_whitelist"] == "peer-1,host:4242"
+ assert data["discovery"]["interface_discovery_blacklist"] == "bad-node,evilentry"
+ assert config["reticulum"]["interface_discovery_whitelist"] == "peer-1,host:4242"
+ assert config["reticulum"]["interface_discovery_blacklist"] == "bad-node,evilentry"
+ assert config.write_called
+
+
+def test_filter_discovered_interfaces_handles_non_list_inputs():
+ result = ReticulumMeshChat.filter_discovered_interfaces(
+ {"unexpected": "shape"},
+ "peer-*",
+ "bad-*",
+ )
+ assert result == {"unexpected": "shape"}
+
+
@pytest.mark.asyncio
async def test_interface_add_includes_discovery_fields(temp_dir):
config = ConfigDict({"reticulum": {}, "interfaces": {}})
diff --git a/tests/backend/test_property_based.py b/tests/backend/test_property_based.py
index 38827969..be9fc966 100644
--- a/tests/backend/test_property_based.py
+++ b/tests/backend/test_property_based.py
@@ -8,6 +8,7 @@ import pytest
from hypothesis import HealthCheck, given, settings
from hypothesis import strategies as st
+from meshchatx.meshchat import ReticulumMeshChat
from meshchatx.src.backend.colour_utils import ColourUtils
from meshchatx.src.backend.identity_manager import IdentityManager
from meshchatx.src.backend.interface_config_parser import InterfaceConfigParser
@@ -317,6 +318,46 @@ def test_interface_config_parser_structured(names, keys, values):
pytest.fail(f"InterfaceConfigParser.parse failed on structured input: {e}")
+@given(
+ interfaces=st.lists(
+ st.dictionaries(
+ keys=st.sampled_from(
+ [
+ "name",
+ "type",
+ "reachable_on",
+ "target_host",
+ "remote",
+ "listen_ip",
+ "port",
+ "target_port",
+ "listen_port",
+ "discovery_hash",
+ "transport_id",
+ "network_id",
+ ]
+ ),
+ values=st.one_of(st.text(), st.integers(), st.none()),
+ max_size=12,
+ ),
+ max_size=40,
+ ),
+ whitelist=st.one_of(st.text(), st.lists(st.text(), max_size=20), st.none()),
+ blacklist=st.one_of(st.text(), st.lists(st.text(), max_size=20), st.none()),
+)
+def test_discovery_filter_robustness(interfaces, whitelist, blacklist):
+ try:
+ filtered = ReticulumMeshChat.filter_discovered_interfaces(
+ interfaces,
+ whitelist,
+ blacklist,
+ )
+ except Exception as e:
+ pytest.fail(f"Discovery filtering crashed: {e}")
+ assert isinstance(filtered, list)
+ assert len(filtered) <= len(interfaces)
+
+
# Strategy for a database message row
st_db_message = st.dictionaries(
keys=st.sampled_from(
@@ -870,3 +911,39 @@ def test_parse_lxmf_display_name_extended(aspect, data):
assert isinstance(result, str)
except Exception:
pass
+
+
+MIN_SIZE_RATIO = 0.2
+
+
+def _is_backup_suspicious_reference(current_stats, baseline):
+ if not baseline:
+ return False
+ prev_count = baseline.get("message_count", 0)
+ prev_bytes = baseline.get("total_bytes", 0)
+ curr_count = current_stats.get("message_count", 0)
+ curr_bytes = current_stats.get("total_bytes", 0)
+ if prev_count > 0 and curr_count == 0:
+ return True
+ if prev_bytes > 100_000 and curr_bytes < prev_bytes * MIN_SIZE_RATIO:
+ return True
+ return False
+
+
+@given(
+ prev_count=st.integers(min_value=0, max_value=1_000_000),
+ prev_bytes=st.integers(min_value=0, max_value=10_000_000),
+ curr_count=st.integers(min_value=0, max_value=1_000_000),
+ curr_bytes=st.integers(min_value=0, max_value=10_000_000),
+)
+@settings(suppress_health_check=[HealthCheck.too_slow])
+def test_is_backup_suspicious_property(prev_count, prev_bytes, curr_count, curr_bytes):
+ from meshchatx.src.backend.database import Database
+
+ baseline = {"message_count": prev_count, "total_bytes": prev_bytes}
+ current_stats = {"message_count": curr_count, "total_bytes": curr_bytes}
+ db = Database(":memory:")
+ db.initialize()
+ result = db._is_backup_suspicious(current_stats, baseline)
+ expected = _is_backup_suspicious_reference(current_stats, baseline)
+ assert result == expected
diff --git a/tests/backend/test_security_fuzzing.py b/tests/backend/test_security_fuzzing.py
index 6dedd909..0a4b952c 100644
--- a/tests/backend/test_security_fuzzing.py
+++ b/tests/backend/test_security_fuzzing.py
@@ -13,6 +13,21 @@ from hypothesis import strategies as st
from meshchatx.meshchat import ReticulumMeshChat
+@given(value=st.one_of(st.text(), st.lists(st.text(), min_size=0, max_size=20)))
+@settings(deadline=None)
+def test_discovery_pattern_sanitization_never_emits_unsafe_tokens(value):
+ patterns = ReticulumMeshChat.sanitize_discovery_patterns(value)
+ assert isinstance(patterns, list)
+ assert len(patterns) <= 128
+ for token in patterns:
+ assert token
+ assert len(token) <= 128
+ assert "," not in token
+ assert "\n" not in token
+ assert "\r" not in token
+ assert token.strip() == token
+
+
@pytest.fixture
def mock_app():
# Save real Identity class to use as base for our mock class
@@ -1280,6 +1295,346 @@ def test_nomadnet_page_content_fuzzing(mock_app, page_content):
convert_nomadnet_field_data_to_map(page_content)
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(path_data=st.one_of(st.text(min_size=0, max_size=10000), st.none()))
+def test_nomadnet_string_data_to_map_fuzzing(mock_app, path_data):
+ """Fuzz NomadNet path variable string parsing (e.g. after backtick in page path)."""
+ from meshchatx.src.backend.nomadnet_utils import convert_nomadnet_string_data_to_map
+
+ result = convert_nomadnet_string_data_to_map(path_data)
+ assert isinstance(result, dict)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ download_id=st.one_of(
+ st.integers(),
+ st.text(min_size=0, max_size=200),
+ st.none(),
+ ),
+)
+def test_nomadnet_download_cancel_fuzzing(mock_app, download_id):
+ """Fuzz nomadnet.download.cancel WebSocket handler."""
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(
+ MagicMock(),
+ {"type": "nomadnet.download.cancel", "download_id": download_id},
+ ),
+ )
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ destination_hash=st.text(min_size=0, max_size=200),
+ page_path=st.text(min_size=0, max_size=2000),
+)
+def test_nomadnet_page_archives_get_fuzzing(mock_app, destination_hash, page_path):
+ """Fuzz nomadnet.page.archives.get WebSocket handler."""
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(
+ MagicMock(),
+ {
+ "type": "nomadnet.page.archives.get",
+ "destination_hash": destination_hash,
+ "page_path": page_path,
+ },
+ ),
+ )
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ archive_id=st.one_of(
+ st.integers(),
+ st.text(min_size=0, max_size=100),
+ st.floats(allow_nan=False, allow_infinity=False),
+ st.none(),
+ ),
+)
+def test_nomadnet_page_archive_load_fuzzing(mock_app, archive_id):
+ """Fuzz nomadnet.page.archive.load and get_archived_page_by_id."""
+ mock_app.database.misc.get_archived_page_by_id(archive_id)
+ import asyncio
+
+ mock_app.database.misc.get_archived_page_by_id.return_value = None
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(
+ MagicMock(),
+ {"type": "nomadnet.page.archive.load", "archive_id": archive_id},
+ ),
+ )
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ destination_hash=st.text(min_size=0, max_size=200),
+ page_path=st.text(min_size=0, max_size=2000),
+ content=st.one_of(
+ st.text(min_size=0, max_size=50000),
+ st.binary(min_size=0, max_size=10000),
+ ),
+)
+def test_nomadnet_page_archive_add_fuzzing(
+ mock_app, destination_hash, page_path, content
+):
+ """Fuzz nomadnet.page.archive.add WebSocket handler and archive_page."""
+ import asyncio
+
+ content_str = (
+ content.decode("utf-8", errors="replace")
+ if isinstance(content, bytes)
+ else content
+ )
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(
+ MagicMock(),
+ {
+ "type": "nomadnet.page.archive.add",
+ "destination_hash": destination_hash,
+ "page_path": page_path,
+ "content": content_str,
+ },
+ ),
+ )
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ destination_hash=st.text(min_size=0, max_size=200),
+ file_path=st.text(min_size=0, max_size=2000),
+)
+def test_nomadnet_file_download_fuzzing(mock_app, destination_hash, file_path):
+ """Fuzz nomadnet.file.download WebSocket handler (path traversal, malformed hash)."""
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ payload = {
+ "type": "nomadnet.file.download",
+ "nomadnet_file_download": {
+ "destination_hash": destination_hash,
+ "file_path": file_path,
+ },
+ }
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(MagicMock(), payload),
+ )
+ except (ValueError, TypeError):
+ pass
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ destination_hash=st.text(min_size=0, max_size=200),
+ page_path=st.text(min_size=0, max_size=2000),
+ field_data=st.one_of(
+ st.dictionaries(
+ keys=st.text(),
+ values=st.one_of(st.text(), st.integers(), st.booleans(), st.none()),
+ ),
+ st.text(min_size=0, max_size=5000),
+ st.lists(st.text(), min_size=0, max_size=20),
+ st.none(),
+ ),
+)
+def test_nomadnet_page_download_fuzzing(
+ mock_app, destination_hash, page_path, field_data
+):
+ """Fuzz nomadnet.page.download WebSocket handler (page_path with backtick, field_data)."""
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ payload = {
+ "type": "nomadnet.page.download",
+ "nomadnet_page_download": {
+ "destination_hash": destination_hash,
+ "page_path": page_path,
+ "field_data": field_data,
+ },
+ }
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(MagicMock(), payload),
+ )
+ except (ValueError, TypeError):
+ pass
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(archive_id=st.one_of(st.integers(), st.text(), st.floats(), st.none()))
+def test_get_archived_page_by_id_fuzzing(mock_app, archive_id):
+ """Fuzz archived page lookup by id (SQL injection, type confusion)."""
+ mock_app.database.misc.get_archived_page_by_id(archive_id)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ forward_to_hash=st.text(min_size=0, max_size=100),
+ identity_hash=st.one_of(st.text(min_size=0, max_size=100), st.none()),
+ source_filter_hash=st.one_of(st.text(min_size=0, max_size=100), st.none()),
+ is_active=st.one_of(st.booleans(), st.integers(), st.none()),
+ name=st.one_of(st.text(min_size=0, max_size=200), st.none()),
+)
+def test_lxmf_forwarding_rule_add_fuzzing(
+ mock_app,
+ forward_to_hash,
+ identity_hash,
+ source_filter_hash,
+ is_active,
+ name,
+):
+ """Fuzz lxmf.forwarding.rule.add WebSocket handler and create_forwarding_rule."""
+ import asyncio
+
+ rule = {"forward_to_hash": forward_to_hash}
+ if identity_hash is not None:
+ rule["identity_hash"] = identity_hash
+ if source_filter_hash is not None:
+ rule["source_filter_hash"] = source_filter_hash
+ if is_active is not None:
+ rule["is_active"] = is_active
+ if name is not None:
+ rule["name"] = name
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(
+ MagicMock(),
+ {"type": "lxmf.forwarding.rule.add", "rule": rule},
+ ),
+ )
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ rule_id=st.one_of(st.integers(), st.text(min_size=0, max_size=50), st.none()),
+)
+def test_lxmf_forwarding_rule_delete_fuzzing(mock_app, rule_id):
+ """Fuzz lxmf.forwarding.rule.delete WebSocket handler."""
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(
+ MagicMock(),
+ {"type": "lxmf.forwarding.rule.delete", "id": rule_id},
+ ),
+ )
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ rule_id=st.one_of(st.integers(), st.text(min_size=0, max_size=50), st.none()),
+)
+def test_lxmf_forwarding_rule_toggle_fuzzing(mock_app, rule_id):
+ """Fuzz lxmf.forwarding.rule.toggle WebSocket handler."""
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(
+ MagicMock(),
+ {"type": "lxmf.forwarding.rule.toggle", "id": rule_id},
+ ),
+ )
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(
+ action=st.text(min_size=0, max_size=200),
+ keys=st.one_of(
+ st.lists(st.text(), min_size=0, max_size=20),
+ st.text(min_size=0, max_size=500),
+ st.integers(),
+ ),
+)
+def test_keyboard_shortcuts_set_fuzzing(mock_app, action, keys):
+ """Fuzz keyboard_shortcuts.set WebSocket handler (action and keys)."""
+ import asyncio
+
+ loop = asyncio.new_event_loop()
+ asyncio.set_event_loop(loop)
+ try:
+ loop.run_until_complete(
+ mock_app.on_websocket_data_received(
+ MagicMock(),
+ {"type": "keyboard_shortcuts.set", "action": action, "keys": keys},
+ ),
+ )
+ except (KeyError, TypeError, ValueError):
+ pass
+ finally:
+ loop.close()
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(message_hash=st.text(min_size=0, max_size=200))
+def test_messages_get_by_hash_fuzzing(mock_app, message_hash):
+ """Fuzz message lookup by hash (SQL, type confusion)."""
+ mock_app.database.messages.get_lxmf_message_by_hash(message_hash)
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(message_hash=st.text(min_size=0, max_size=200))
+def test_messages_delete_by_hash_fuzzing(mock_app, message_hash):
+ """Fuzz single message delete by hash."""
+ try:
+ mock_app.database.messages.delete_lxmf_message_by_hash(message_hash)
+ except Exception:
+ pass
+
+
+@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
+@given(message_hashes=st.lists(st.text(min_size=0, max_size=100), min_size=0, max_size=50))
+def test_messages_delete_by_hashes_fuzzing(mock_app, message_hashes):
+ """Fuzz bulk message delete by hashes."""
+ try:
+ mock_app.database.messages.delete_lxmf_messages_by_hashes(message_hashes)
+ except Exception:
+ pass
+
+
@settings(suppress_health_check=[HealthCheck.function_scoped_fixture], deadline=None)
@given(
table_name=st.text(min_size=0, max_size=100),
diff --git a/tests/backend/test_startup_advanced.py b/tests/backend/test_startup_advanced.py
index 045ae8a9..d61f96ee 100644
--- a/tests/backend/test_startup_advanced.py
+++ b/tests/backend/test_startup_advanced.py
@@ -160,7 +160,11 @@ def test_database_integrity_recovery(mock_rns, temp_dir):
patch("meshchatx.src.backend.identity_context.RNProbeHandler"),
patch("meshchatx.src.backend.identity_context.TranslatorHandler"),
patch("meshchatx.src.backend.identity_context.CommunityInterfacesManager"),
+ patch(
+ "meshchatx.src.backend.identity_context.IntegrityManager",
+ ) as mock_im_class,
):
+ mock_im_class.return_value.check_integrity.return_value = (True, [])
mock_db_instance = mock_db_class.return_value
# Fail the first initialize call
mock_db_instance.initialize.side_effect = [
@@ -228,7 +232,64 @@ def test_identity_loading_fallback(mock_rns, temp_dir):
mock_gen_id.get_private_key.assert_called()
-# 4. Test flags/envs
+# 4. Database health issues set on setup and exposed to app
+def test_database_health_issues_set_on_setup(mock_rns, temp_dir):
+ with (
+ patch("meshchatx.src.backend.identity_context.Database") as mock_db_class,
+ patch(
+ "meshchatx.src.backend.identity_context.ConfigManager",
+ ) as mock_config_class,
+ patch("meshchatx.src.backend.identity_context.MessageHandler"),
+ patch("meshchatx.src.backend.identity_context.AnnounceManager"),
+ patch("meshchatx.src.backend.identity_context.ArchiverManager"),
+ patch("meshchatx.src.backend.identity_context.MapManager"),
+ patch("meshchatx.src.backend.identity_context.DocsManager"),
+ patch("meshchatx.src.backend.identity_context.NomadNetworkManager"),
+ patch("meshchatx.src.backend.identity_context.TelephoneManager"),
+ patch("meshchatx.src.backend.identity_context.VoicemailManager"),
+ patch("meshchatx.src.backend.identity_context.RingtoneManager"),
+ patch("meshchatx.src.backend.identity_context.RNCPHandler"),
+ patch("meshchatx.src.backend.identity_context.RNStatusHandler"),
+ patch("meshchatx.src.backend.identity_context.RNProbeHandler"),
+ patch("meshchatx.src.backend.identity_context.TranslatorHandler"),
+ patch("meshchatx.src.backend.identity_context.CommunityInterfacesManager"),
+ patch(
+ "meshchatx.src.backend.identity_context.IntegrityManager",
+ ) as mock_int_class,
+ patch("aiohttp.web.run_app"),
+ ):
+ mock_int_class.return_value.check_integrity.return_value = (True, [])
+ mock_db_instance = mock_db_class.return_value
+ mock_db_instance.check_db_health_at_open.return_value = [
+ "Database content anomaly: test."
+ ]
+ mock_config = mock_config_class.return_value
+ mock_config.auth_session_secret.get.return_value = base64.urlsafe_b64encode(
+ secrets.token_bytes(32),
+ ).decode()
+ mock_config.display_name.get.return_value = "Test"
+ mock_config.lxmf_propagation_node_stamp_cost.get.return_value = 0
+ mock_config.lxmf_delivery_transfer_limit_in_bytes.get.return_value = 1000000
+ mock_config.lxmf_inbound_stamp_cost.get.return_value = 0
+ mock_config.lxmf_preferred_propagation_node_destination_hash.get.return_value = None
+ mock_config.lxmf_local_propagation_node_enabled.get.return_value = False
+ mock_config.libretranslate_url.get.return_value = "http://localhost:5000"
+ mock_config.translator_enabled.get.return_value = False
+ mock_config.initial_docs_download_attempted.get.return_value = True
+
+ app = ReticulumMeshChat(
+ identity=mock_rns["id_instance"],
+ storage_dir=temp_dir,
+ reticulum_config_dir=temp_dir,
+ )
+ app.run(host="127.0.0.1", port=8000, launch_browser=False, enable_https=False)
+ assert getattr(app, "database_health_issues", []) == [
+ "Database content anomaly: test."
+ ]
+ app.teardown_identity()
+
+
+# 5. Test flags/envs
def test_cli_flags_and_envs(mock_rns, temp_dir):
with (
patch("meshchatx.meshchat.ReticulumMeshChat") as mock_app_class,
diff --git a/tests/frontend/InterfacesDiscoveryActions.test.js b/tests/frontend/InterfacesDiscoveryActions.test.js
new file mode 100644
index 00000000..48fd8c6e
--- /dev/null
+++ b/tests/frontend/InterfacesDiscoveryActions.test.js
@@ -0,0 +1,172 @@
+import { mount } from "@vue/test-utils";
+import { describe, it, expect, vi, beforeEach } from "vitest";
+import InterfacesPage from "../../meshchatx/src/frontend/components/interfaces/InterfacesPage.vue";
+
+vi.mock("../../meshchatx/src/frontend/js/GlobalState", () => ({
+ default: {
+ config: { theme: "light" },
+ hasPendingInterfaceChanges: false,
+ modifiedInterfaceNames: new Set(),
+ },
+}));
+
+vi.mock("../../meshchatx/src/frontend/js/Utils", () => ({
+ default: {
+ formatBytes: (b) => `${b} B`,
+ isInterfaceEnabled: () => true,
+ },
+}));
+
+vi.mock("../../meshchatx/src/frontend/js/ToastUtils", () => ({
+ default: {
+ success: vi.fn(),
+ error: vi.fn(),
+ },
+}));
+
+vi.mock("../../meshchatx/src/frontend/js/ElectronUtils", () => ({
+ default: {
+ relaunch: vi.fn(),
+ isElectron: () => false,
+ },
+}));
+
+const mockAxios = {
+ get: vi.fn((url) => {
+ if (url.includes("/api/v1/reticulum/interfaces")) {
+ return Promise.resolve({ data: { interfaces: {} } });
+ }
+ if (url.includes("/api/v1/app/info")) {
+ return Promise.resolve({ data: { app_info: { is_reticulum_running: true } } });
+ }
+ if (url.includes("/api/v1/interface-stats")) {
+ return Promise.resolve({ data: { interface_stats: { interfaces: [] } } });
+ }
+ if (url.includes("/api/v1/reticulum/discovery")) {
+ return Promise.resolve({
+ data: {
+ discovery: {
+ discover_interfaces: true,
+ interface_discovery_whitelist: "",
+ interface_discovery_blacklist: "",
+ },
+ },
+ });
+ }
+ if (url.includes("/api/v1/reticulum/discovered-interfaces")) {
+ return Promise.resolve({ data: { interfaces: [], active: [] } });
+ }
+ return Promise.resolve({ data: {} });
+ }),
+ post: vi.fn(() => Promise.resolve({ data: {} })),
+ patch: vi.fn(() => Promise.resolve({ data: {} })),
+};
+window.axios = mockAxios;
+
+describe("InterfacesPage discovery actions", () => {
+ beforeEach(() => {
+ vi.clearAllMocks();
+ });
+
+ it("shows discovery action menu and allows announce", async () => {
+ const wrapper = mount(InterfacesPage, {
+ global: {
+ stubs: {
+ RouterLink: true,
+ MaterialDesignIcon: true,
+ Toggle: true,
+ ImportInterfacesModal: true,
+ Interface: true,
+ },
+ mocks: {
+ $t: (key) => key,
+ $router: { push: vi.fn() },
+ },
+ },
+ });
+
+ await wrapper.setData({
+ discoveredInterfaces: [
+ {
+ name: "Peer Node",
+ type: "TCPClientInterface",
+ reachable_on: "10.0.0.8",
+ port: 4242,
+ discovery_hash: "hash-1",
+ },
+ ],
+ discoveryConfig: {
+ discover_interfaces: true,
+ interface_discovery_sources: "",
+ interface_discovery_whitelist: "",
+ interface_discovery_blacklist: "old-entry",
+ required_discovery_value: null,
+ autoconnect_discovered_interfaces: 0,
+ network_identity: "",
+ },
+ });
+
+ const actionButton = wrapper.find('button[title="Discovery actions"]');
+ expect(actionButton.exists()).toBe(true);
+ await actionButton.trigger("click");
+ expect(wrapper.text()).toContain("Allow this announce");
+
+ const allowButton = wrapper.findAll("button").find((btn) => btn.text().includes("Allow this announce"));
+ expect(allowButton).toBeTruthy();
+ await allowButton.trigger("click");
+
+ expect(mockAxios.patch).toHaveBeenCalledWith("/api/v1/reticulum/discovery", {
+ interface_discovery_whitelist: "10.0.0.8:4242",
+ interface_discovery_blacklist: null,
+ });
+ });
+
+ it("blacklists announce and removes same token from whitelist", async () => {
+ const wrapper = mount(InterfacesPage, {
+ global: {
+ stubs: {
+ RouterLink: true,
+ MaterialDesignIcon: true,
+ Toggle: true,
+ ImportInterfacesModal: true,
+ Interface: true,
+ },
+ mocks: {
+ $t: (key) => key,
+ $router: { push: vi.fn() },
+ },
+ },
+ });
+
+ await wrapper.setData({
+ discoveredInterfaces: [
+ {
+ name: "Peer Node",
+ type: "TCPClientInterface",
+ reachable_on: "10.0.0.8",
+ port: 4242,
+ discovery_hash: "hash-2",
+ },
+ ],
+ discoveryConfig: {
+ discover_interfaces: true,
+ interface_discovery_sources: "",
+ interface_discovery_whitelist: "10.0.0.8:4242,other",
+ interface_discovery_blacklist: "",
+ required_discovery_value: null,
+ autoconnect_discovered_interfaces: 0,
+ network_identity: "",
+ },
+ });
+
+ await wrapper.find('button[title="Discovery actions"]').trigger("click");
+ const blockButton = wrapper.findAll("button").find((btn) => btn.text().includes("Blacklist this announce"));
+ expect(blockButton).toBeTruthy();
+ await blockButton.trigger("click");
+
+ expect(mockAxios.patch).toHaveBeenCalledWith("/api/v1/reticulum/discovery", {
+ interface_discovery_whitelist: null,
+ interface_discovery_blacklist: "10.0.0.8:4242",
+ });
+ });
+});
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────